package com.leigq.www.consumer.consumer;

import com.leigq.www.common.constant.QueueConstant;
import com.leigq.www.consumer.config.KafKaConsumerConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Headers;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Stream;

import static com.leigq.www.common.constant.QueueConstant.BZI_TYPE_RECORD_HEADER_KEY;

/**
 * 订单消费客户端
 *
 * @author leiguoqing
 */
@Component
@Slf4j
public class OrderConsumer {

    /*↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ 全注解方式 实现消息队列（非延迟） ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓*/

    /**
     * 订单消费队列，手动签收
     * <br/>
     * 俩个同组的 Customer 订阅同一个主题，主题中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。
     * <br/>
     * <code>@KafkaListener</code>: 配置监听的哪一个主题，同时在没有主题的情况下会去创建
     */
    @KafkaListener(topics = {QueueConstant.ORDER_TOPIC_NAME}, groupId = QueueConstant.ORDER_GROUP_NAME, containerFactory = "errorListenerContainerFactory")
    public void onOrderMessage1(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("简单消费：" + record.topic() + "-" + record.key() + "-" + record.partition() + "-" + record.value());

        // 手动签收 ACK, 确认一条消息已经被消费
        ack.acknowledge();

        try {
            // 模拟异常，放开注释即可正常消费
            // int a = 1 / 0;

            // 这里实现实体的业务逻辑，多个消费者处理是需注意并发问题（加锁）
            log.warn("onOrderMessage1 执行了相关业务....");

            log.warn("onOrderMessage1 消费完成....");
        } catch (Exception e) {
            // 主动抛异常触发重试及投放到消费错误队列
            log.error("异常信息：", e);
            throw new RuntimeException("消费异常");
        }
    }

    /**
     * 订单队列批量消费，自动签收
     * <br/>
     * 俩个同组的 Customer 订阅同一个主题，主题中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。
     * <br/>
     * <code>@KafkaListener</code>: 配置监听的哪一个主题，同时在没有主题的情况下会去创建
     * <code>orderBatchFactory</code>: 订单批量消费者工厂，{@link KafKaConsumerConfig#orderBatchFactory(KafkaProperties)}
     */
    @KafkaListener(topics = {QueueConstant.ORDER_TOPIC_NAME}, groupId = QueueConstant.ORDER_GROUP_NAME, containerFactory = "orderBatchFactory")
    public void onOrderMessage2(List<ConsumerRecord<?, ?>> records) {
        System.out.println(">>>批量消费一次，records.size()=" + records.size());
        // 消费的哪个topic、partition的消息,打印出消息内容
        records.forEach(record -> {
            System.out.println("批量消费：" + record.topic() + "-" + record.key() + "-" + record.partition() + "-" + record.value());
        });
    }

    /**
     * 订单手动签收消息，手动签收
     * <br/>
     * 俩个同组的 Customer 订阅同一个主题，主题中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。
     * <br/>
     * <code>@KafkaListener</code>: 配置监听的哪一个主题，同时在没有主题的情况下会去创建
     * <code>orderManualFactory</code>: 订单手动签收消息工厂，{@link KafKaConsumerConfig#orderManualFactory(KafkaProperties)}
     */
    @KafkaListener(topics = {QueueConstant.ORDER_TOPIC_NAME}, groupId = QueueConstant.ORDER_GROUP_NAME, containerFactory = "orderManualFactory")
    public void onOrderMessage3(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("订单手动签收消息：" + record.topic() + "-" + record.key() + "-" + record.partition() + "-" + record.value());
        ack.acknowledge();
    }

    /*↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ 全注解方式 实现消息队列（非延迟） ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑*/

    /**
     * 消费错误队列
     * <br/>
     * <code>@KafkaListener</code>: 配置监听的哪一个主题，同时在没有主题的情况下会去创建
     */
    @KafkaListener(topics = {QueueConstant.ERROR_ORDER_TOPIC_NAME}, groupId = QueueConstant.ERROR_GROUP_NAME)
    public void onErrorMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println("错误队列消费：" + record.topic() + "-" + record.key() + "-" + record.partition() + "-" + record.value());
        // 可将消费错误的队列进行持久化，做后续处理（在headers里面添加biz-type可进行业务区分）
        final Headers headers = record.headers();

        final AtomicReference<String> bizTypeAtomic = new AtomicReference<>();

        Stream.of(headers.toArray()).filter(header -> StringUtils.equals(header.key(), BZI_TYPE_RECORD_HEADER_KEY)).findFirst().ifPresent(h -> {
            bizTypeAtomic.set(new String(h.value(), StandardCharsets.UTF_8));
        });

        final String bizType = bizTypeAtomic.get();
        System.out.println("业务类型 = " + bizType);
        ack.acknowledge();
    }
}
